package com.atguigu.app.dws;

import com.alibaba.fastjson.JSON;
import com.alibaba.fastjson.JSONObject;
import com.atguigu.app.func.DimAsyncFunction;
import com.atguigu.bean.DwsTradeOrderStatusStaticBean;
import com.atguigu.util.DateFormatUtil;
import com.atguigu.util.KafkaUtil;
import com.atguigu.util.MyClickHouseUtil;
import org.apache.flink.api.common.eventtime.WatermarkStrategy;
import org.apache.flink.api.common.functions.FlatMapFunction;
import org.apache.flink.api.common.functions.ReduceFunction;
import org.apache.flink.api.common.functions.RichFlatMapFunction;
import org.apache.flink.api.common.state.StateTtlConfig;
import org.apache.flink.api.common.state.ValueState;
import org.apache.flink.api.common.state.ValueStateDescriptor;
import org.apache.flink.api.common.time.Time;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.datastream.AsyncDataStream;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.datastream.KeyedStream;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.windowing.WindowFunction;
import org.apache.flink.streaming.api.windowing.assigners.TumblingEventTimeWindows;
import org.apache.flink.streaming.api.windowing.windows.TimeWindow;
import org.apache.flink.util.Collector;

import java.time.Duration;
import java.util.concurrent.TimeUnit;

/**
 * @author shkstart
 * @creat 2022/9/4-16:41
 */
public class DwsTradeOrderStatusStaticWindow {
    public static void main(String[] args) throws Exception {
        //todo 1.获取环境
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.setParallelism(1);
        //todo 2.设置后端
        //todo 3.读取Kafka中DWD层数据
        String groupId = "dws_trade_order_status_static_window";
        String topicName = "dwd_trade_order_detail";
        DataStreamSource<String> source = env.addSource(KafkaUtil.getKafkaConsumer(topicName, groupId));
        //todo 4.转换数据结构，去重得到独立用户
        KeyedStream<JSONObject, String> keyedStream = source.flatMap(new FlatMapFunction<String, JSONObject>() {
            @Override
            public void flatMap(String value, Collector<JSONObject> out) throws Exception {
                try {
                    JSONObject jsonObject = JSONObject.parseObject(value);
                    out.collect(jsonObject);
                } catch (Exception e) {
                }
            }
        }).keyBy(s -> s.getString("user_id"));
//        KeyedStream<String, String> keyedStream = source.keyBy(value -> JSON.parseObject(value).getString("user_id"));

        SingleOutputStreamOperator<DwsTradeOrderStatusStaticBean> flatMap = keyedStream.flatMap(new RichFlatMapFunction<JSONObject, DwsTradeOrderStatusStaticBean>() {
            //获取状态
            ValueState<String> state;

            @Override
            public void open(Configuration parameters) throws Exception {
                ValueStateDescriptor<String> last_dt = new ValueStateDescriptor<>("last_dt", String.class);
                last_dt.enableTimeToLive(new StateTtlConfig.Builder(Time.days(1)).setUpdateType(StateTtlConfig.UpdateType.OnCreateAndWrite).build());
                state = getRuntimeContext().getState(last_dt);
            }

            @Override
            public void flatMap(JSONObject jsonObject, Collector<DwsTradeOrderStatusStaticBean> out) throws Exception {
//                JSONObject jsonObject = JSON.parseObject(value);
                String dt = DateFormatUtil.toDate(jsonObject.getLong("od_ts") * 1000L);
                ValueState<String> last_dt = this.state;
                //如果是独立访客，就将下单次数与下单人数均设置为1L
                if (last_dt == null || !dt.equals(last_dt)) {
                    DwsTradeOrderStatusStaticBean result = DwsTradeOrderStatusStaticBean.builder()
                            .courseId(jsonObject.getString("course_id"))
                            .courseName(jsonObject.getString("course_name"))
                            .countOrder(1L)
                            .sumPerson(1L)
                            .sumAmount(jsonObject.getBigDecimal("final_amount"))
                            .ts(jsonObject.getLong("od_ts") * 1000L)
                            .build();
                    out.collect(result);
                    //如果不是独立，就将下单次数设置为1L，下单人数设置为0L
                } else {
                    DwsTradeOrderStatusStaticBean result = DwsTradeOrderStatusStaticBean.builder()
                            .courseId(jsonObject.getString("course_id"))
                            .courseName(jsonObject.getString("course_name"))
                            .countOrder(1L)
                            .sumPerson(0L)
                            .sumAmount(jsonObject.getBigDecimal("final_amount"))
                            .ts(jsonObject.getLong("od_ts") * 1000L)
                            .build();
                    out.collect(result);

                }
            }
        });
        //todo 5.添加水位线
        SingleOutputStreamOperator<DwsTradeOrderStatusStaticBean> andWatermarks = flatMap.assignTimestampsAndWatermarks(WatermarkStrategy
                .<DwsTradeOrderStatusStaticBean>forBoundedOutOfOrderness(Duration.ofSeconds(2L))
                .withTimestampAssigner((x, y) -> x.getTs()));
        //todo 6.按课程id分组聚合
        SingleOutputStreamOperator<DwsTradeOrderStatusStaticBean> reduce = andWatermarks.keyBy(value -> value.getCourseName())
                .window(TumblingEventTimeWindows.of(org.apache.flink.streaming.api.windowing.time.Time.seconds(10L)))
                .reduce(new ReduceFunction<DwsTradeOrderStatusStaticBean>() {
                            @Override
                            public DwsTradeOrderStatusStaticBean reduce(DwsTradeOrderStatusStaticBean value1, DwsTradeOrderStatusStaticBean value2) throws Exception {
                                value1.setCountOrder(value1.getCountOrder() + value2.getCountOrder());
                                value1.setSumPerson(value1.getSumPerson() + value2.getSumPerson());
                                value1.setSumAmount(value1.getSumAmount().add(value2.getSumAmount()));
                                return value1;
                            }
                        },
                        new WindowFunction<DwsTradeOrderStatusStaticBean, DwsTradeOrderStatusStaticBean, String, TimeWindow>() {
                            @Override
                            public void apply(String s, TimeWindow window, Iterable<DwsTradeOrderStatusStaticBean> input, Collector<DwsTradeOrderStatusStaticBean> out) throws Exception {
                                DwsTradeOrderStatusStaticBean value = input.iterator().next();
                                value.setStt(DateFormatUtil.toYmdHms(window.getStart()));
                                value.setEdt(DateFormatUtil.toYmdHms(window.getEnd()));
                                value.setTs(System.currentTimeMillis());
                                out.collect(value);
                            }
                        }
                );
        //todo 7.1查询关联课程维度获取学科ID
        String tableName1 = "dim_base_category_info".toUpperCase();
        SingleOutputStreamOperator<DwsTradeOrderStatusStaticBean> categoryBeanStream =
                AsyncDataStream.unorderedWait(reduce, new DimAsyncFunction<DwsTradeOrderStatusStaticBean>(tableName1) {
                            @Override
                            public String getKey(DwsTradeOrderStatusStaticBean input) {
                                return input.getCourseId();
                            }

                            @Override
                            public void join(DwsTradeOrderStatusStaticBean input, JSONObject jsonObject) {
                                input.setSubjectId(jsonObject.getString("subjectId"));
                            }
                        }, 5,
                        TimeUnit.MINUTES);

        //todo 7.2查询关联学科维度
        String tableName2 = "dim_base_subject_info".toUpperCase();
        SingleOutputStreamOperator<DwsTradeOrderStatusStaticBean> subjectBeanStream =
                AsyncDataStream.unorderedWait(categoryBeanStream, new DimAsyncFunction<DwsTradeOrderStatusStaticBean>(tableName2) {
            @Override
            public String getKey(DwsTradeOrderStatusStaticBean input) {
                return input.getSubjectId();
            }

            @Override
            public void join(DwsTradeOrderStatusStaticBean input, JSONObject jsonObject) {
                input.setSubjectName(jsonObject.getString("subjectName"));
                input.setCategoryId(jsonObject.getString("categoryId"));
            }
        }, 5, TimeUnit.MINUTES);
        //todo 8.查询关联类别维度
        String tableName3 = "dim_base_category_info";
        SingleOutputStreamOperator<DwsTradeOrderStatusStaticBean> result = AsyncDataStream.unorderedWait(subjectBeanStream,
                new DimAsyncFunction<DwsTradeOrderStatusStaticBean>(tableName3) {
                    @Override
                    public String getKey(DwsTradeOrderStatusStaticBean input) {
                        return input.getCategoryId();
                    }

                    @Override
                    public void join(DwsTradeOrderStatusStaticBean input, JSONObject jsonObject) {
                        input.setCategoryName(jsonObject.getString("categoryName"));
                    }
                }, 5, TimeUnit.MINUTES);
        //todo 9.写出CK
        result.print("result>>>>");
        result.addSink(MyClickHouseUtil.getSink("insert into dws_trade_order_status_static_window values(" +
                "?,?,?,?,?,?," +
                "?,?,?,?,?,?" +
                ")"));
        //todo 10.执行
        env.execute(groupId);
    }
}
